package cn.zk.consumer;

import cn.zk.UserDto;
import cn.zk.config.RabbitMQConfig;
import cn.zk.domain.MqMessage;
import cn.zk.mapper.MqMessageMapper;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.User;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者
 */
@Slf4j
@Component
public class ReceiveHandler {
    @Resource
    private MqMessageMapper mqMessageMapper;
    @Resource
    private MessageConverter messageConverter;
    @Resource(name = "mqWorkers")
    private ExecutorService executorService;
    @Resource
    private AtomicCount atomicCount;

    //监听email队列
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_EMAIL})
    public void receive_email(Message message, Channel channel) throws IOException {
        log.info("收到消息:message={},channel={}",message,channel);
        UserDto user = (UserDto)messageConverter.fromMessage(message);
        log.info("消息转化:user={}",user);

        // 判断是否已经消费过消息，避免消息的重复消费
//        String messageId = message.getMessageProperties().getMessageId();
//        if(isConsume(messageId)){
//            // 调试用
//            System.err.println(messageId+"已经消费过!");
//            return;
//        }

        // 处理业务
//        System.err.println("receive_email:"+new String(message.getBody(),"UTF-8"));

        // 将消息存入db中标记已经消费过
//        flagMassage(message, messageId);

        // 制造异常 导致消息重复消费
//        int i = 1/0;
        executorService.submit(()->{
            // 模拟处理业务
            try {
                TimeUnit.SECONDS.sleep(10);
                atomicCount.inc(user.getId());
            } catch (InterruptedException e) {
                log.error("线程睡眠失败",e);
            }
            // 手动签收
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            } catch (IOException e) {
                log.error("签收失败",e);
            }
        });

    }

    @Bean("mqWorkers")
    public ExecutorService threadPoolHandler() {
       return Executors.newFixedThreadPool(10);
    }


    //监听sms队列
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_SMS})
    public void receive_sms(Message message, Channel channel) throws IOException {
        // 判断是否已经消费过消息，避免消息的重复消费
        String messageId = message.getMessageProperties().getMessageId();
        if(isConsume(messageId)){
            // 调试用
            System.err.println(messageId+"已经消费过!");
            return;
        }

        // 处理业务
        System.err.println("receive_sms:"+new String(message.getBody(),"UTF-8"));

        // 手动签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

        // 将消息存入db中标记已经消费过
        flagMassage(message, messageId);

    }

    /**
     * 消息是否存在
     * @param MessageId
     * @return
     */
    boolean isConsume(String MessageId){
        MqMessage message = mqMessageMapper.findById(MessageId);
        return !(message == null);
    }

    /**
     * 标记已经消费的消息
     * @param message
     * @param messageId
     */
    @Transactional
    public void flagMassage(Message message, String messageId) {
        MqMessage mqMessage = new MqMessage();
        mqMessage.setMessageId(messageId);
        mqMessage.setTime(new Date());
        mqMessage.setQueueName(message.getMessageProperties().getConsumerQueue());
        mqMessageMapper.add(mqMessage);
    }

} 